/*
  gconn.c
  Description: Connection Pool for GIGI
  Created: 2009.10.19
  Version: 1.0
*/


/* cannot use WINDOWS here, because not include config.h at this point */

#if defined(_WIN32) || defined(__WIN64)
   #define FD_SETSIZE 128
#endif

#ifdef HAVE_CONFIG_H
  #include "config.h"
#endif

#include "gconn.h"
#include <stdio.h>
#include <assert.h>

#define THIS_FILE_ERROR_ID  40000

#ifdef WINDOWS

  #define GIGI_GET_LAST_FD_ERROR()  WSAGetLastError()

  int g_init_windows_socket = GIGI_FALSE;

  #define INIT_WIN_SOCKET()                                 \
         do { if (g_init_windows_socket == GIGI_FALSE )     \
         { WSADATA wsa; g_init_windows_socket = GIGI_TRUE;  \
           WSAStartup (MAKEWORD(1, 1), &wsa); } } while (0)

#else

  #define GIGI_GET_LAST_FD_ERROR()  errno
  #define INIT_WIN_SOCKET() do {} while(0)

#endif

static void _gigi_event_free (gigi_event* p);

static gigi_event * _gigi_event_new (FD_SOCKET helpFD);

static void _socket_list_insert_order (GIGI_SOCKET* root, GIGI_SOCKET fd);

static void _socket_list_free (GIGI_SOCKET fd);

int _socket_list_del (GIGI_SOCKET* root, GIGI_SOCKET fd);

static GIGI_SOCKET * _socket_list_by_fd (GIGI_SOCKET* root, FD_SOCKET fd);

static void _socket_free (GIGI_SOCKET fd);

static int _socket_register (GIGI_SOCKET fd, GIGIONE context, GIGI_ID nodeId);

static void _socket_unregister (GIGI_SOCKET fd);

static int _gigi_shakehand (GIGI_SOCKET fd, int * cmd,int noACK);

static int _gigi_socket_handle (GIGIONE context, fd_set * r,
                                FD_SOCKET nMinFD, FD_SOCKET nMaxFD);

static int _gigi_select_prepare (GIGIONE context, fd_set * fs, struct timeval * tv,
                                 FD_SOCKET * nMinFD, FD_SOCKET * nMaxFD);

static fd_pool * _gigi_find_pool (GIGIONE context, GIGI_ID nodeId, int create);

static GIGI_SERVICE _gigi_find_service (GIGIONE context, GIGI_ID service_id, int lock);

static gigi_event* _gigi_alloc_event (GIGIONE context);

static int _gigi_idle (GIGIONE context);

static int _gigi_init_context (GIGIONE context, char ** conn, int port,
                               const char * token_string);

static int _gigi_free_objects (GIGIONE context);

static int _gigi_shutdown (GIGIONE context);

static int _gigi_startup (GIGIONE context);

static void _gigi_callback (GIGIONE context, int msg, int lParam, int wParam);

static int _gigi_service_handle (void * ctx, GIGI_SOCKET fd);

static void _gigi_clean_invalid_socket (GIGIONE context);

static GIGI_SERVICE gigi_open_service_internal (
    GIGIONE context, GIGI_ID service_id, int backlog,
    GIGI_SERVICE_FUNC func, void * ctx);


#ifdef GIGI_DEBUG

static void _gigi_print_socket (const char * s, GIGI_SOCKET fd)
{
  char sFromNode[64],sToNode[64],sServID[64];

  if (fd)
  {
    printf ("%s (%d) connect=%s->%s service=%s create=%s fd=%d\n",
      s, 
      THREAD_ID (), 
      node_id_to_string ( fd->_created ? fd->_node_pool->_node_id
                                       : fd->_node_pool->_context->_my_node_id,
                                       sFromNode ),

      node_id_to_string ( fd->_created ? fd->_node_pool->_context->_my_node_id
                                       : fd->_node_pool->_node_id,
                                       sToNode ),

      service_id_to_string (fd->_service_id, sServID), 
      fd->_created ? "yes" : "no" , 
      fd->_tcp_fd 
    );
  }
  else
  {
    printf ("%s (%d) socket error\n", s, THREAD_ID ());
  }
};

#endif


/* function _gigi_event_new
 *
 */
static gigi_event * _gigi_event_new (FD_SOCKET helpFD)
{
  gigi_event * evt;
  
  evt = NEW_OBJECT (gigi_event);
  
  assert ( evt );

  if (evt == NULL)
  {
    LOG_OUT_OF_MEMORY();
    return NULL;
  }

  if ( 0 != fd_socket_pipe (helpFD, &evt->_in, &evt->_out) )
  {
    free (evt);
    evt = NULL;
  }

  return evt;
};

/* function _gigi_event_free
 *
 */
static void _gigi_event_free (gigi_event * root)
{
  while (root)
  {
    gigi_event* g = root;
    LIST_NEXT (root);
    FD_CLOSE_SOCKET (g->_in);
    FD_CLOSE_SOCKET (g->_out);
    free (g);
  }
};

/* function _gigi_alloc_event
 *
 */
static gigi_event * _gigi_alloc_event (GIGIONE context)
{
  gigi_event * p;

  if (context->_evt_pool)
  {
    LIST_POP_FRONT (context->_evt_pool, p);
  }
  else
  {
    p = _gigi_event_new ( INVALID_SOCKET );

    if (p == NULL)
    {
      return NULL;
    }
  }

  p->_next = NULL;

  return p;
};

/* function : _socket_list_insert_order
 *
 */
static void _socket_list_insert_order (GIGI_SOCKET* root, GIGI_SOCKET fd)
{
  GIGI_SOCKET r;
  int order;

  r = *root;
  order = fd->_order;

  if (r == NULL || order < r->_order)
  {
    fd->_next = r;
    *root = fd;

    return;
  }

  while ( r->_next && r->_next->_order < order)
  {
    LIST_NEXT (r);
  };

  fd->_next = r->_next;
  r->_next = fd;

#ifdef GIGI_DEBUG
#if    0
  for (r = *root; r && r->_next; LIST_NEXT (r) )
  {
    assert ( r->_order < r->_next->_order);
  }
#endif
#endif

};

/* function _socket_list_free
 * 
 */
static void _socket_list_free (GIGI_SOCKET fd)
{
  GIGI_SOCKET s;

  while (fd)
  {
    LIST_POP_FRONT (fd, s);
    
    _socket_unregister(s);

    shutdown ( s->_tcp_fd._fd, 2);
    
    FD_CLOSE_SOCKET (s->_tcp_fd._fd);
    
    free (s);
  }
};

/* function _socket_register
 *
 */
static int _socket_register (GIGI_SOCKET fd, GIGIONE context, GIGI_ID node_id)
{
  fd_pool * pool;

  if ( fd == NULL || context == NULL)
  {
    return GIGI_E_INTERNAL_ERROR;
  }
  
  MUTEX_LOCK (context->_lock);

  pool = _gigi_find_pool (context, node_id, GIGI_TRUE);

  if (pool)
  {
    fd->_node_pool = pool;
    pool->_ref_count++;
  }

  MUTEX_UNLOCK (context->_lock);

  return pool ? GIGI_OK : GIGI_E_OUT_OF_MEMORY;
}

/* function _socket_unregister
 *
 */
static void _socket_unregister (GIGI_SOCKET fd)
{
  int nDel;
  fd_pool * pool;

  if (fd == NULL || fd->_node_pool == NULL)
  {
    return;
  }

  nDel = GIGI_FALSE;
  pool = fd->_node_pool;

  MUTEX_LOCK (pool->_context->_lock);
  
  fd->_node_pool = NULL;
  pool->_ref_count--;

  if (pool->_ref_count <= 0)
  {
    RING_DEL (pool);
    nDel = GIGI_TRUE;
  }

  MUTEX_UNLOCK (pool->_context->_lock);

  if (nDel)
  {
    assert ( pool->_pool_count == 0 );
    assert ( pool->_peer_socket == NULL );
    assert ( pool->_socket_pool == NULL );
    free (pool);
  }
}

/* function _socket_list_del
 *
 */

int _socket_list_del (GIGI_SOCKET * root, GIGI_SOCKET fd)
{
  int del;

  for (;*root; LIST_PP_NEXT (root) )
  {
    if ( *root == fd ) break;
  }
  
  del = *root != NULL;

  if (del)
  {
    LIST_PP_DEL (root);
  }

  return del;
};

/* function _socket_list_by_fd
 *
 */
static GIGI_SOCKET * _socket_list_by_fd (GIGI_SOCKET * root, FD_SOCKET fd)
{
  for (;*root; LIST_PP_NEXT (root) )
  {
    if ( (*root)->_tcp_fd._fd == fd)
    {
      return root;
    }
  }

  return NULL;
};

/* function _socket_free
 *
 */
static void _socket_free (GIGI_SOCKET fd)
{
  if (fd) 
  {
    _socket_unregister(fd);

    shutdown (fd->_tcp_fd._fd, 2);
    closesocket (fd->_tcp_fd._fd);
    free (fd); 
  }
}

/* function _gigi_get_sockets_from_cache 
 *
 */
static GIGI_SOCKET _gigi_get_sockets_from_cache (GIGIONE context)
{
  GIGI_SOCKET fd = NULL, s;             
  int i;
  
  MUTEX_LOCK (context->_lock);

  for ( i = 0; i < NODE_HASH_COUNT; i++)
  {
    fd_pool* pI   = RING_BEGIN ( context->_fd_pool_hash + i, fd_pool*);
    fd_pool* pEND  = RING_END  ( context->_fd_pool_hash + i, fd_pool*);

        for (; pI != pEND; RING_NEXT (pI) )
    {
      while ( pI->_peer_socket )
      {
        LIST_POP_FRONT (pI->_peer_socket, s);
        LIST_PUSH_FRONT ( fd, s);
      }

      while ( pI->_socket_pool )
      {
        LIST_POP_FRONT (pI->_socket_pool, s);
        LIST_PUSH_FRONT (fd, s);
      }

      pI->_pool_count = 0;
    }
  };
    context->_conn_count = 0;

  if ( RING_IS_EMPTY (&context->_service_list) == GIGI_FALSE )
  {
    GIGI_SERVICE pI   = RING_BEGIN ( &context->_service_list, GIGI_SERVICE);
    GIGI_SERVICE pEND   = RING_END ( &context->_service_list, GIGI_SERVICE);

    for ( ; pI != pEND; RING_NEXT (pI) )
    {
      while ( pI->_accept_list )
      {
        LIST_POP_FRONT (pI->_accept_list, s);
        LIST_PUSH_FRONT (fd, s);
      }
    };
  }

  MUTEX_UNLOCK ( context->_lock );

  return fd;
};

/* function _gigi_find_pool
 *
 */
static fd_pool * _gigi_find_pool (GIGIONE context, GIGI_ID node_id, int create)
{
  ring_root * root;
  fd_pool * r, * rEND;

  root = context->_fd_pool_hash + ( (unsigned)( node_id % NODE_HASH_COUNT) );

  for ( r = RING_BEGIN (root,fd_pool*), rEND = RING_END(root,fd_pool*); 
        r != rEND; RING_NEXT (r) )
  {
    if ( r->_node_id == node_id) 
    {
      return r;
    }
  };

  if (!create)
  {
    return NULL;
  }

  r = NEW_OBJECT (fd_pool);
  
  if ( r == NULL)
  {
    LOG_OUT_OF_MEMORY();
    return NULL;
  }

  RING_PUSH_FRONT (root, r);
  r->_context            = context;
  r->_ref_count      = 0;
  r->_node_id        = node_id;
  r->_pool_count     = 0;
  r->_socket_pool    = NULL;
  r->_peer_socket    = NULL;
  r->_active_peer_socket_order = 0;

  return r;
}

/* function _gigi_select_prepare
 *
 */
static int _gigi_select_prepare (GIGIONE context, fd_set * fs, 
                                 struct timeval * tv,
                                 FD_SOCKET * nMinFD, FD_SOCKET * nMaxFD)
{
  int nRet = GIGI_OK;
  int hash_ind;
  fd_pool *pI,*pEND;
  GIGI_SOCKET sI;

  tv;

  FD_FDSET ( context->_my_fd           , fs, *nMinFD, *nMaxFD);
  FD_FDSET ( context->_mgr_event->_out, fs, *nMinFD, *nMaxFD);


  MUTEX_LOCK (context->_lock);

  memset (context->_peer_cache, 0, sizeof (context->_peer_cache));

  for (hash_ind = 0; hash_ind < NODE_HASH_COUNT; hash_ind++)
  for (pI = RING_BEGIN(context->_fd_pool_hash + hash_ind,fd_pool*),
    pEND=RING_END(context->_fd_pool_hash + hash_ind, fd_pool*); pI != pEND; RING_NEXT (pI) )
  {
    // only peek the first 2 into select list
    //
    int nCount = 2;

    pI->_active_peer_socket_order = 0;

    for (sI = pI->_peer_socket; sI && nCount; LIST_NEXT (sI), nCount-- )
    {
      unsigned nHashID = ( (unsigned)sI->_tcp_fd._fd) % PEER_SOCKET_CACHE;

      FD_FDSET (sI->_tcp_fd._fd, fs, *nMinFD, *nMaxFD);

      nRet++;
      context->_peer_cache[nHashID] = sI;
      pI->_active_peer_socket_order = sI->_order;
    }
  };

  MUTEX_UNLOCK (context->_lock);

#ifdef WINDOWS
  assert( fs->fd_count < (FD_SETSIZE -4) );
#endif

  return nRet;
};

/* function _gigi_find_service
 *
 */
static GIGI_SERVICE _gigi_find_service (GIGIONE context, GIGI_ID service_id, int lock)
{
  GIGI_SERVICE i,iEND,ret = NULL;
  
  if (lock) MUTEX_LOCK (context->_lock );

  i = RING_BEGIN ( &(context->_service_list), GIGI_SERVICE);
  iEND = RING_END   ( &(context->_service_list), GIGI_SERVICE);

  for ( ;i!=iEND; RING_NEXT (i) )
  {
    if ( i->_service_id == service_id)
    {
      ret = i;
      break;
    }
  }

  if (lock) MUTEX_UNLOCK (context->_lock);

  return ret;
};

/* function _gigi_init_context
 *
 */
static int _gigi_init_context (GIGIONE context, char ** conn, int port,
                               const char* token_string)
{
  int i;

  // init the variables
  //
  MUTEX_INIT (context->_lock);

  RING_INIT ( & (context->_service_list) );
  RING_INIT ( & (context->_callback_list) );
  RING_INIT ( & (context->_func_list_root) );

  for (i = 0; i < NODE_HASH_COUNT; i++)
  {
    RING_INIT ( context->_fd_pool_hash + i );
  }

  token_string = conn_string_get (conn, "token", token_string);

  context->_last_service_id   = 1;

  // read config
  //
  context->_cfg_pool_max       = conn_string_get_int (conn, "conn_pool_max",
                                               2, 2, 64);

  context->_gigi_port         = conn_string_get_int (conn, "port",
                                               port, 0, 0xFFFF);

  context->_conn_expried_time  = conn_string_get_int (conn, "conn-expried",
                                               10, 3, 0x7fffffff);

  context->_conn_max_count     = conn_string_get_int (conn, "max-count-of-pool",
                                               DEF_MAX_COUNT_OF_POOL, 10, 100);
  
  // set up the socket
  //
  context->_my_fd = fd_tcp_bind (&context->_gigi_port);

  if ( FD_SOCKET_VALID (context->_my_fd) == GIGI_FALSE )
  {
    return GIGI_E_BIND_FAILED;
  }

  // init the event object
  //
  for (i=0;i<EVENT_INIT_COUNT;i++)
  {
    gigi_event* evt = _gigi_event_new (context->_my_fd);
    
    if (evt == NULL)
    {
      continue;
    }
    
    LIST_PUSH_FRONT (context->_evt_pool, evt);
  }

  context->_mgr_event = _gigi_alloc_event (context);

  if (context->_mgr_event == NULL)
  {
    FD_CLOSE_SOCKET (context->_my_fd);
    _gigi_event_free (context->_evt_pool);
    context->_evt_pool = NULL;

    return GIGI_E_INTERNAL_ERROR;
  }
  
  // generate the MyID
  //
  {
    int isAddr;
    unsigned uIP = fd_gethostid (&isAddr);

    if (!isAddr)
    {
      unsigned uCfgIP = conn_string_get_ip (conn,"IP",NULL);
      
      if (uCfgIP != INADDR_NONE)
      {
        uIP = uCfgIP;
        isAddr = GIGI_TRUE;
      }
    }

    context->_my_node_id = GIGI_ID_MAKE (uIP,context->_gigi_port,0);
  };

  context->_gigi_service = gigi_open_service_internal (
    context, 
    GIGI_SERVICE_GIGI ,
    30, 
    _gigi_service_handle, 
    context
  );

  return GIGI_OK;
};

/* function _gigi_free_objects
 *
 */
static int _gigi_free_objects (GIGIONE context)
{
  int i, hash_ind;

  if ( context->_gigi_service )
  {
    gigi_close_service (context->_gigi_service);
    context->_gigi_service = NULL;
  };

  _socket_list_free ( _gigi_get_sockets_from_cache (context ) );

  MUTEX_LOCK (context->_lock);

  // node_pool
  //  
  for (hash_ind = 0; hash_ind < NODE_HASH_COUNT; hash_ind++)
  {
#if 0
   if ( RING_IS_EMPTY (context->_fd_pool_hash + hash_ind)  == 0)
   {
     char ss[64];
     fd_pool* p = (fd_pool*) ( context->_fd_pool_hash[hash_ind]._next );
     printf ("un del %d, %s, refCount = %d\n",hash_ind, node_id_to_string (p->_node_id,ss), p->_ref_count );
   }
#else
   assert ( RING_IS_EMPTY (context->_fd_pool_hash + hash_ind)  );
#endif
  }

  RING_INIT ( & (context->_service_list) );

  if (context->_mgr_event)
  {
    LIST_PUSH_FRONT (context->_evt_pool, context->_mgr_event);
    context->_mgr_event = NULL;
  }

  _gigi_event_free (context->_evt_pool);
  context->_evt_pool = NULL;

  // free socket
  FD_CLOSE_SOCKET (context->_my_fd);

  for (i = 0; i<PEER_SOCKET_CACHE; i++)
  {
    context->_peer_cache[i] = NULL;
  };

  while (RING_IS_EMPTY (&(context->_func_list_root)) == GIGI_FALSE)
  {
    ring_root_ptr func = RING_BEGIN(&(context->_func_list_root), ring_root_ptr);
    RING_DEL(func);
    free (func);
  }

  MUTEX_UNLOCK (context->_lock);

  MUTEX_FREE (context->_lock);

  return GIGI_OK;
};

/* function _gigi_on_recv
 *
 */
static int _gigi_on_recv (GIGIONE context, GIGI_SOCKET fd)
{
  acceptEvent* pEvt = NULL;
  int shakehandCmd;
  GIGI_SERVICE pService;
  char chACK = GIGI_OK;
  int nRet = _gigi_shakehand (fd, &shakehandCmd, GIGI_TRUE);
  
  if ( nRet != GIGI_OK )
  {
    _socket_free (fd);
    
    assert (nRet == GIGI_E_CLOSED_BY_PEER || nRet == GIGI_E_NO_SERVICE);

    if (nRet == GIGI_E_CLOSED_BY_PEER || nRet == GIGI_E_NO_SERVICE)
    {
      nRet = GIGI_OK;
    };

    return nRet;
  }

  MUTEX_LOCK (context->_lock);

  pService = _gigi_find_service (context, fd->_service_id, GIGI_FALSE);

  if (pService == NULL)
  {
    GIGI_SERVICE root = (GIGI_SERVICE)& (context->_service_list);
    
    GIGI_ERROR (
      "Service (%d) not found, %d, %s\n", 
      (int)fd->_service_id, 
      (int)root->_next->_service_id, 
      RING_IS_EMPTY (root) ? "empty" : "not empty"
    );

    nRet = GIGI_E_NO_SERVICE;
  goto Cleanup;
  }

  if (pService->_func)
  {
    MUTEX_UNLOCK (context->_lock);

  nRet = fd_send ( fd->_tcp_fd._fd, &chACK, sizeof (chACK), -1 );

  if (nRet >= 0 )
  {
    nRet = pService->_func (pService->_func_ctx, fd);
  }

    gigi_closesocket (fd, nRet < 0 ? 1: 0 );

    return GIGI_OK;
  }

  if (RING_IS_EMPTY (&pService->_wait_list) == GIGI_FALSE )
  {
    pEvt = (acceptEvent*) (pService->_wait_list._next);
    RING_DEL (pEvt);
    RING_INIT (pEvt);
    pEvt->_result = fd;    
  }
  else if (pService->_accept_list_count < pService->_accept_list_max)
  {
    fd->_next = NULL;
    
    if (pService->_accept_list)
    {
      GIGI_SOCKET lst = pService->_accept_list;
      for (;lst->_next; LIST_NEXT (lst))
      {
      };

      lst->_next = fd;
    }
    else
    {
      pService->_accept_list = fd;
    }

    pService->_accept_list_count++;
  } 
  else
  {
    GIGI_ERROR( "accept List(%d/%d)\n", 
      pService->_accept_list_count,
      pService->_accept_list_max
    );
    
  nRet = GIGI_E_SERVER_BUSY;
  goto Cleanup;
  }

Cleanup:;
  MUTEX_UNLOCK (context->_lock);

  chACK = (char)nRet;
  fd_send (fd->_tcp_fd._fd, &chACK, sizeof(chACK), -1);

  if (nRet != GIGI_OK)
  {
    _socket_free (fd);
  };

  if (pEvt)
  {
    _gigi_event_set (pEvt->_event);
  }

  return nRet;
};

/* function _gigi_socket_on_accept
 *
 */
static void _gigi_socket_on_accept (GIGIONE context)
{
  char Buffer[16];
  int n;
  struct sockaddr_in addr;
  int addrlen = sizeof (addr);
  GIGI_SOCKET fd;
  GIGI_ID nodeAddr;

  // step 1: create socket object
  //
  fd = NEW_OBJECT (gigi_socket_context);

  if (fd == NULL)
  {
    LOG_OUT_OF_MEMORY ();
    return;
  }

  memset (fd, 0, sizeof (*fd) );
  fd_stream_init (fd->_tcp_fd );

  // step 2: accept
  //
  fd->_tcp_fd._fd = accept (context->_my_fd, (struct sockaddr*)&addr, (socklen_t*)&addrlen);
  n = FD_SOCKET_VALID ( fd->_tcp_fd._fd ) ? GIGI_OK : GIGI_E_INTERNAL_ERROR;
  CER (n);

  n = fd_recv ( fd->_tcp_fd._fd, Buffer, sizeof (Buffer), SHAKEHAND_TIMEOUT );
  CER (n);

  GIGI_GET_ID    ( Buffer + 0, nodeAddr );
  GIGI_GET_INT32 ( Buffer + 8, fd->_order);
  GIGI_GET_INT32 ( Buffer + 12,fd->_tcp_fd._bo);

  GIGI_PUT_INT32 (Buffer + 0, BO_MY );
  n = fd_send ( fd->_tcp_fd._fd, Buffer, sizeof (int), SHAKEHAND_TIMEOUT);
  CER (n);

  // step 3:  
  //
  fd->_raw_mode = (char)( nodeAddr == 0 );
  n = _socket_register (fd, context, nodeAddr);
  CER (n);

  n = _gigi_on_recv (context, fd);

Cleanup:;

  /*this return just for cubrid to detect whether cub_cluster is running*/
  if(n == GIGI_E_CLOSED_BY_PEER)
  {
    return ;
  }

  if ( n < 0 )
  {
    GIGI_ERROR ("accept socket error (%u)", -n);
  }
}

/* function _gigi_socket_on_reuse
 *
 */
static void _gigi_socket_on_reuse (GIGIONE context, fd_set * r, FD_SOCKET nMinFD,
                                   FD_SOCKET nMaxFD)
{
  GIGI_SOCKET sHeader = NULL, sI = NULL, fd, *ifd;
  unsigned nCacheID;
  fd_pool* iNodeInfo;
  size_t i;
  int n;

  MUTEX_LOCK (context->_lock);

#ifdef WINDOWS
  nMinFD,nMaxFD;
  for (i = 0; i < r->fd_count ; i++)
  {  
    FD_SOCKET fdSocket = r->fd_array[i];
#else
  for (i = nMinFD; i<= nMaxFD ; i++)
  {
    FD_SOCKET fdSocket = i;
    if (FD_ISSET (i, r) == GIGI_FALSE) continue;
#endif
    nCacheID = ( (unsigned)fdSocket) % PEER_SOCKET_CACHE;
    ifd = NULL;

    // first lookup in Cache
    //
    if ( context->_peer_cache[nCacheID] && context->_peer_cache[nCacheID]->_tcp_fd._fd == fdSocket)
    {
      iNodeInfo = context->_peer_cache[nCacheID]->_node_pool;
      context->_peer_cache[nCacheID] = NULL;
      ifd = _socket_list_by_fd (&iNodeInfo->_peer_socket, fdSocket);
      assert ( ifd );
#ifdef GIGI_DEBUG
      context->_cacheHit ++;
#endif
    }
    else
    {
      int hash_ind;
    fd_pool* pEND;

      for (hash_ind = 0; hash_ind < NODE_HASH_COUNT; hash_ind++)
    {
        pEND      = RING_END   (context->_fd_pool_hash + hash_ind, fd_pool*);
      iNodeInfo = RING_BEGIN (context->_fd_pool_hash + hash_ind, fd_pool*);
      
    for (;iNodeInfo != pEND && ifd == NULL; RING_NEXT (iNodeInfo))
    {
          ifd = _socket_list_by_fd (&iNodeInfo->_peer_socket, fdSocket);
        }
    }

#ifdef GIGI_DEBUG
      context->_cacheMissing++;
#endif
    }

    if (ifd == NULL)
    {
      assert (0);
      continue;
    }

    LIST_POP_FRONT (*ifd, fd);
    fd->_next = NULL;
    context->_conn_count--;
    
    if (sHeader)
    {
      sI->_next = fd; 
      sI = sI->_next;
    } 
    else
    {
      sHeader = sI = fd;
    }
  }

  MUTEX_UNLOCK (context->_lock);
  
  while (sHeader)
  {
    LIST_POP_FRONT (sHeader, fd);
    
    fd->_next = NULL;

    n = _gigi_on_recv (context, fd);

    assert( n >= GIGI_OK );
  }
};

/* function _gigi_socket_handle
 *
 */
static int _gigi_socket_handle (GIGIONE context, fd_set * r, FD_SOCKET nMinFD,
                                FD_SOCKET nMaxFD)
{
  if ( FD_ISSET (context->_my_fd, r) )
  {
    //GIGI_TRACE ("recv new connection", "");
    _gigi_socket_on_accept (context);
    FD_CLR (context->_my_fd, r );
  }
  
  if (FD_ISSET (context->_mgr_event->_out, r))
  {
    //GIGI_TRACE ("new reuse candidate join", "");

    _gigi_event_wait (context->_mgr_event, 1000);
    FD_CLR (context->_mgr_event->_out, r);
  }

#ifdef WINDOWS
  if ( r->fd_count > 0 )
  {
    //GIGI_TRACE (
    //  "reuse connection %d(%d)", 
    //  r->fd_array[0], 
    //  r->fd_count
    //);
#else
  {
#endif

    _gigi_socket_on_reuse (context, r, nMinFD, nMaxFD);
  }

  return GIGI_OK;
};

/* function _gigi_idle
 *
 */
static int _gigi_idle (GIGIONE context)
{
  _gigi_clean_invalid_socket (context);
  context->_idle_count++;

  _gigi_callback (context, GIGI_CB_IN_IDLE, 0, 0);
  
  return GIGI_OK;
};

/* function _gigi_shakehand
 *
 */
static int _gigi_shakehand (GIGI_SOCKET fd, int * cmd, int noACK)
{
  int  n;
  char Buffer[12];
  char chACK = GIGI_OK;

  GIGI_PERF_BEGIN

  if (fd->_created)
  {
    GIGI_PUT_INT32 ( Buffer + 0, *cmd );
    GIGI_PUT_ID    ( Buffer + 4, fd->_service_id);

    n = fd_send (fd->_tcp_fd._fd, Buffer, sizeof (Buffer), SHAKEHAND_TIMEOUT);
    
    if ( n > 0)
    {
      n = fd_recv (fd->_tcp_fd._fd, &chACK, sizeof (char), SHAKEHAND_TIMEOUT);
    }
  }
  else
  {
    n = fd_recv (fd->_tcp_fd._fd, Buffer, sizeof (Buffer), SHAKEHAND_TIMEOUT);

    if ( n > 0)
    {      
      chACK = GIGI_OK;
    
      GIGI_GET_INT32 ( Buffer + 0, *cmd );
      GIGI_GET_ID    ( Buffer + 4, fd->_service_id);
      
      if (*cmd == SHAKEHAND_CMD_CONNECT )
      {
        chACK = _gigi_find_service (fd->_node_pool->_context, fd->_service_id, GIGI_TRUE) ? GIGI_OK : GIGI_E_NO_SERVICE;
      }
    
    if ( chACK != GIGI_OK || noACK == GIGI_FALSE)
    {
      n = fd_send (fd->_tcp_fd._fd, &chACK, sizeof (char), SHAKEHAND_TIMEOUT);
    }
    }
  }

  if ( n > 0 && chACK != GIGI_OK )
  {
    n = (int)chACK;
  }

  fd->_error = n > 0 ? 0 : 1;

  GIGI_PERF_CHECK (50)

  return n < 0 ? n : GIGI_OK;
};

/* function gigi_socket_create
 *
 */
GIGI_SOCKET gigi_socket_create (GIGI_ID node_id, GIGI_ID service_id,
                                GIGI_ID myId, GIGIONE context, int * nErr)
{
  static int nNextOrder = 1;
  GIGI_SOCKET fd = NEW_OBJECT (gigi_socket_context);
  char Buffer[16 + 12];
  int n;

  if (fd == NULL)
  {
    LOG_OUT_OF_MEMORY();
  *nErr = GIGI_E_OUT_OF_MEMORY;
    return fd;
  };

  INIT_WIN_SOCKET();

  memset (fd, 0, sizeof (*fd) );

  fd->_created   = GIGI_TRUE;
  fd->_order     = nNextOrder++;
  fd->_service_id = service_id;
  fd->_raw_mode   = (char)( myId == 0 );
  fd_stream_init ( fd->_tcp_fd );

  fd->_tcp_fd._fd = fd_connect ( GIGI_ID_IP(node_id), GIGI_ID_PORT(node_id), CONNECT_TIMEOUT);
  
  if ( FD_SOCKET_VALID (fd->_tcp_fd._fd) == GIGI_FALSE)
  {
    *nErr = GIGI_E_CONNECT_ERROR;
    free (fd);
    return NULL;
  }

  GIGI_PUT_ID    ( Buffer + 0, myId      );
  GIGI_PUT_INT32 ( Buffer + 8, fd->_order);
  GIGI_PUT_INT32 ( Buffer + 12,BO_MY     );
  // for shakehand 
  GIGI_PUT_INT32 ( Buffer + 16,SHAKEHAND_CMD_CONNECT );
  GIGI_PUT_ID    ( Buffer + 20,service_id );

  n = fd_send (fd->_tcp_fd._fd, Buffer, sizeof (Buffer), SHAKEHAND_TIMEOUT);
  CER (n);

  n = fd_recv (fd->_tcp_fd._fd, Buffer, sizeof (int) + sizeof(char)   , SHAKEHAND_TIMEOUT);
  CER (n);

  *nErr = (int) Buffer[4];
  GIGI_GET_INT32 ( Buffer + 0, fd->_tcp_fd._bo );
  CER(*nErr);

  if (context)
  {
    n = _socket_register(fd, context, node_id);
    CER (n);
  }

Cleanup:;
  if ( n < 0 )
  {
    *nErr = n;
  }

  if (*nErr != GIGI_OK )
  {
    _socket_free (fd);
    fd = NULL;
  }

  return fd;
};

/* function gigi_socket
 *
 */
GIGI_SOCKET gigi_socket (GIGIONE context, GIGI_ID node_id, GIGI_ID service_id,
                         int raw_mode, int * err_code)
{
  GIGI_SOCKET fd = NULL;
  fd_pool* pNodePool = NULL;
  int shakehandCmd;
  int nErrTmp;

  if (NULL == err_code)
  {
    err_code = &nErrTmp;
  }

  *err_code = GIGI_OK;
  
  GIGI_PERF_BEGIN

  if ( raw_mode )
  {
    fd = gigi_socket_create (node_id, service_id, 0, NULL, err_code);
    goto Cleanup;
  }

  GIGI_HANDLE_CHECK (NULL);

  // first get it from nodehash
  //
  while ( GIGI_TRUE )
  {
    MUTEX_LOCK (context->_lock);

    pNodePool = _gigi_find_pool (context, node_id, GIGI_FALSE);
  
    if (pNodePool && pNodePool->_socket_pool)
    {
      LIST_POP_FRONT ( pNodePool->_socket_pool, fd);
      pNodePool->_pool_count--;
      fd->_next = NULL;
    }

    MUTEX_UNLOCK (context->_lock);

    if (fd == NULL)
    {
      break;
    }

    fd->_service_id = service_id;

    shakehandCmd = SHAKEHAND_CMD_CONNECT;
    *err_code = _gigi_shakehand (fd, &shakehandCmd, 0);
  
    if (*err_code == GIGI_OK)
    {
      goto Cleanup;
    }

    _socket_free (fd);
  fd = NULL;

    if (*err_code == GIGI_E_NO_SERVICE)
    {
      goto Cleanup;
    }
  }

  fd = gigi_socket_create (node_id, service_id, gigi_my_id (context), context,err_code );

Cleanup:;

  GIGI_PERF_CHECK (50)
  return fd;
};

/* function gigi_closesocket
 *
 */
int gigi_closesocket (GIGI_SOCKET fd, int forceclose)
{
  GIGIONE context;
  int nRet = GIGI_OK, shakehandCmd;
  GIGI_PERF_BEGIN

  if (fd == NULL)
  {
    return GIGI_E_INVALID_HANDLE;
  };

  //forceclose = GIGI_TRUE;

  if (fd->_error || forceclose || fd->_raw_mode)
  {
    //_gigi_print_socket ("shakehand error", fd);
    _socket_free (fd);
    return nRet;
  }

  context = fd->_node_pool->_context;

  //_gigi_print_socket ("shakehand begin", fd);

  if (fd->_created)
  {
    // put into conn nodehash
    //  
    MUTEX_LOCK (context->_lock);

    if (fd->_node_pool->_pool_count < context->_cfg_pool_max)
    {
      _socket_list_insert_order (&fd->_node_pool->_socket_pool, fd);
      fd->_node_pool->_pool_count++;
    }
    else
    {
      forceclose = GIGI_TRUE;
    }

    MUTEX_UNLOCK (context->_lock);

    shakehandCmd = forceclose ? SHAKEHAND_CMD_CLOSE : SHAKEHAND_CMD_REUSE;

    // if shakehand failed
    //
    nRet = _gigi_shakehand (fd, &shakehandCmd, 0);

    if ( GIGI_OK != nRet && forceclose == GIGI_FALSE)
    {
      MUTEX_LOCK (context->_lock);

      if (_socket_list_del (& (fd->_node_pool->_socket_pool), fd))
      {
        fd->_node_pool->_pool_count--;
      }
      else
      {
        assert (0);
      }

      MUTEX_UNLOCK (context->_lock);

      forceclose = GIGI_TRUE;
    }

    if (forceclose)
    {
      _socket_free (fd);
    }
  }
  else
  {
    nRet = _gigi_shakehand (fd, &shakehandCmd, 0);
    
    if (nRet != GIGI_OK || shakehandCmd == SHAKEHAND_CMD_CLOSE )
    {
      _socket_free (fd);
    }
    else
    {
      int NotifyEvent;

      MUTEX_LOCK (context->_lock);

      NotifyEvent = fd->_order < fd->_node_pool->_active_peer_socket_order || 
                  fd->_node_pool->_active_peer_socket_order == 0;
      fd->_last_used_time = (int)time(NULL);

      if(context->_conn_count < context->_conn_max_count)
      {
        _socket_list_insert_order (&fd->_node_pool->_peer_socket, fd );
        context->_conn_count++;
      }
      else
      {
        forceclose = GIGI_TRUE;
      }

      MUTEX_UNLOCK (context->_lock);

      if (NotifyEvent)
      {
        _gigi_event_set (context->_mgr_event);
      }

      if (forceclose)
      {
        _socket_free (fd);
      }

    }
  }
  
  GIGI_PERF_CHECK (50)

  return nRet;
}

/* function _gigi_accept_internal
 *
 */
static GIGI_SOCKET _gigi_accept_internal (GIGIONE context, GIGI_ID * from, GIGI_ID service_id,
                                 acceptEvent * evt, int timeout)
{
  GIGI_SERVICE pServ;
  GIGI_SOCKET fd = NULL;

  evt->_event = NULL;
  evt->_result= NULL;

  MUTEX_LOCK (context->_lock);

  pServ = _gigi_find_service (context, service_id, GIGI_FALSE);

  if (pServ)
  {
    if (pServ->_accept_list)
    {
      LIST_POP_FRONT (pServ->_accept_list, fd);
      pServ->_accept_list_count--;
    }

    if (fd == NULL)
    {
      evt->_result = NULL;    
      evt->_event = _gigi_alloc_event (context);

      if (evt->_event)
      {
        RING_PUSH_BACK ( (acceptEvent*)& (pServ->_wait_list), evt);
      }
      else
      {
        assert (0);
      }
    };
  }

  MUTEX_UNLOCK (context->_lock);

  if (fd)
  {
    if (from)
    {
      *from = fd->_node_pool->_node_id;
    }

    fd->_next = NULL;

    return fd;
  }

  if (evt->_event == NULL)
  {
    return NULL;
  };

  _gigi_event_wait (evt->_event, timeout);

  MUTEX_LOCK (context->_lock);

  if (NULL == evt->_result)
  {
    RING_DEL (evt);
    RING_INIT (evt);
  };

  LIST_PUSH_FRONT (context->_evt_pool, evt->_event);

  MUTEX_UNLOCK (context->_lock);

  if (evt->_result && from)
  {
    *from = evt->_result->_node_pool->_node_id;
  }

  return evt->_result;
};


/* function gigi_accept
 *
 */
GIGI_SOCKET gigi_accept (GIGIONE context, GIGI_ID * from,
                         GIGI_ID service_id, int timeout)
{
  acceptEvent evt;

  GIGI_HANDLE_CHECK (NULL);

  return _gigi_accept_internal (context, from, service_id, &evt, timeout);
};

/* function _gigi_callback
 *
 */
static void _gigi_callback (GIGIONE context, int msg, int lParam, int wParam)
{
  gigi_callback_info *i, *iEND;

  i   = RING_BEGIN ( &context->_callback_list, gigi_callback_info*);
  iEND= RING_END ( &context->_callback_list, gigi_callback_info*);

  for (;i!=iEND;RING_NEXT (i))
  {
     (*i->_func) (i->_ctx, msg, lParam, wParam);
  };
};

// thread

GIGIONE gigi_global_context = NULL;

/* function gigi_create
 *
 */
GIGIONE gigi_create (GIGI_LPCSTR conn_string, GIGI_LPCSTR token_string,
                     int port, int * errcode)
{
  int nErr = GIGI_OK;
  GIGI_LPSTR * connArgv = NULL;
  GIGIONE This = NULL;

#ifdef GIGI_CONN_STRING
  {
    GIGI_LPCSTR connV[3], newConn;

    connV[0] = conn_string ? conn_string : "";
    connV[1] = getenv(GIGI_CONN_STRING);
    connV[2] = NULL;
    
    newConn = strings_join (NULL, 0, connV, -1, ',',0);  
    
    if (newConn)
    {
      connArgv = conn_string_parser (newConn,',');
      free ( (void *)newConn);
    }
  }

#else
  connArgv = conn_string_parse (conn_string ? conn_string : "", ',');
#endif

  if (connArgv == NULL)
  {
    GIGI_ERROR ("invalid conn string %s", conn_string);
    return NULL;
  }

  INIT_WIN_SOCKET();

  This = NEW_OBJECT (gigi_context);
  
  if (This == NULL)
  {
    free (connArgv);

    LOG_OUT_OF_MEMORY();

    return NULL;
  }
  
  memset (This, 0, sizeof (gigi_context));

  nErr = _gigi_init_context (This, connArgv, port, token_string);

  if ( nErr == GIGI_OK)
  {
    nErr = _gigi_startup (This);
  }

  if (nErr != GIGI_OK)
  {
    GIGI_ERROR("gigi_init error %d\n", -nErr);
    gigi_free (This);
    This = NULL;
  }
  else
  {
    if (gigi_global_context == NULL)
    {
      gigi_global_context = This;
    }
  }

  free (connArgv);

  return This;
};

/* function gigi_free
 *
 */
int gigi_free (GIGIONE context)
{
  int Err = GIGI_OK;

  GIGI_HANDLE_CHECK (GIGI_E_INVALID_HANDLE);

  Err = _gigi_shutdown (context);

  _gigi_free_objects (context);

  free (context);

  if (context == gigi_global_context)
  {
    gigi_global_context = NULL;
  }

  return Err;
}

/* function gigi_deamon
 *
 */
#ifdef WINDOWS
DWORD __stdcall _gigi_daemon (void * p)
#else
void * _gigi_daemon (void * p)
#endif
{
  GIGIONE This = (GIGIONE)p;
  struct timeval tv;
  GIGI_TIME  lastIdle = 0, now;
  fd_set fdr;
  int n;

  while ( This->_is_run )
  {
    FD_SOCKET nMinFD = 0xFFFFFFF;
    FD_SOCKET nMaxFD = 0;

    tv.tv_sec = SELECT_TIMEOUT;
    tv.tv_usec= 0;

    FD_ZERO (&fdr);

    _gigi_select_prepare (This, &fdr, &tv, &nMinFD, &nMaxFD);

#ifdef WINDOWS
    n = select (0, &fdr, 0, 0, &tv);
#else
    n = select (nMaxFD + 1, &fdr, 0, 0, &tv);
    if (n > 0) 
    {
      int nErrNo,slen = sizeof (nErrNo);
      getsockopt(nMaxFD, SOL_SOCKET, SO_ERROR, (void *)&nErrNo, &slen);
      if (nErrNo != 0)
      {
        n = 0;
      }
    }
#endif
    if (This->_is_run == GIGI_FALSE)
    {
      break;
    }

    if (n == 0)
    {
      // go through
    }
    else if (n>0)
    {
      _gigi_socket_handle (This, &fdr, nMinFD, nMaxFD);
    }
    else
    {
      GIGI_ERROR ("Socket Fatal Error (%d)", GIGI_GET_LAST_FD_ERROR ());
      break;
    }

    now = gigi_now();
    if ( (lastIdle + IDLE_INTERNAL) < now )
    {
      lastIdle = now;
      _gigi_idle(This);
    }
  }  // end of while (This->_m_is_run)

  //GIGI_TRACE ("Deamon thread exit",0);

  This->_is_run = GIGI_FALSE;

#ifdef _WIN32
  return 0;
#else
  return NULL;
#endif
};

/* function _gigi_startup
 *
 */
int _gigi_startup (GIGIONE context)
{
  unsigned long dwID;

  assert ( context && context->_is_run == GIGI_FALSE && context->_thread == THREAD_NULL);

  if (context == NULL)
  {
    return GIGI_E_INTERNAL_ERROR;
  };

  context->_is_run = GIGI_TRUE;
  THREAD_BEGIN (context->_thread, _gigi_daemon, context, dwID);

  if (context->_thread == THREAD_NULL)
  {
    context->_is_run = GIGI_FALSE;
    return GIGI_E_THREAD_ERROR;
  }

  return GIGI_OK;
}

/* function _gigi_shutdown
 *
 */
static int _gigi_shutdown (GIGIONE context)
{
  int Err = GIGI_OK;

  if (context->_thread == THREAD_NULL)
  {
    return Err;
  }

  // set exit flag
  context->_is_run = GIGI_FALSE;

  // notify the thread to check the exit flag
  //
  _gigi_event_set (context->_mgr_event);

  if ( THREAD_JOIN (context->_thread) == GIGI_FALSE )
  {
    Err = GIGI_E_THREAD_ERROR;
  }

  THREAD_CLOSE (context->_thread);

  return Err;
}

/* function gigi_my_id
 *
 */
GIGI_ID gigi_my_id (GIGIONE context)
{
  GIGI_HANDLE_CHECK (0);
  return context->_my_node_id;
}

/* function gigi_open_service_internal
 *
 */
static GIGI_SERVICE gigi_open_service_internal (
    GIGIONE context, GIGI_ID service_id, int backlog,
    GIGI_SERVICE_FUNC func, void * ctx)
{
  GIGI_SERVICE serv;
  
  GIGI_HANDLE_CHECK (NULL);

  serv = NEW_OBJECT (struct gigi_service_context_t);

  if (serv == NULL)
  {
    LOG_OUT_OF_MEMORY ();
    return NULL;
  }

  MUTEX_LOCK (context->_lock);

  if (_gigi_find_service (context, service_id, GIGI_FALSE) )
  {
    free (serv);
    serv = NULL;
  }
  else
  {
    serv->_service_id = service_id;
    serv->_context = context;
    RING_INIT ( &serv->_wait_list );
    serv->_accept_list = NULL;
    serv->_accept_list_count = 0;
    serv->_accept_list_max = backlog < 2 ? 2 : (backlog >64  ? 64 : backlog);
    serv->_func = func;
    serv->_func_ctx = ctx;

    RING_PUSH_BACK ( (GIGI_SERVICE)& (context->_service_list), serv );
  }

  MUTEX_UNLOCK (context->_lock);

  return serv;
};

/* function gigi_open_service
 *
 */
GIGI_SERVICE gigi_open_service (GIGIONE context, GIGI_ID service_id, int backlog)
{
  return gigi_open_service_internal (context, service_id, backlog, NULL, NULL);
};


/* function gigi_close_service
 *
 */
int gigi_close_service (GIGI_SERVICE serv)
{
  acceptEvent * evtRoot, * pEvt;

  if (serv == NULL)
  {
    return GIGI_E_INTERNAL_ERROR;
  };

  /* remove from serv list */
  MUTEX_LOCK (serv->_context->_lock);

  RING_DEL (serv);
  RING_INIT (serv);

  MUTEX_UNLOCK (serv->_context->_lock);


  /* release the acceptor */
  evtRoot = (acceptEvent*)( &(serv->_wait_list) );
    
  while ( RING_IS_EMPTY(evtRoot) == GIGI_FALSE)
  {
    pEvt = evtRoot->_next;
    pEvt->_result = NULL;

    RING_DEL (pEvt);
    RING_INIT (pEvt);

    _gigi_event_set( pEvt->_event );
  }

  /* close the socket in wait list */
  _socket_list_free (serv->_accept_list );

  free (serv);

  return GIGI_OK;
};

/* function _gigi_service_handle
 *
 */
static int _gigi_service_handle (void * ctx, GIGI_SOCKET fd)
{
  int nCMD;

  if ( gigi_recv_object (fd, GIGI_TYPE_INT32, &nCMD, NULL) < 0 )
  {
    return GIGI_E_INTERNAL_ERROR;
  }

  switch (nCMD)
  {
  case GIGI_SERVICE_CMD_GETID:
    {
      GIGI_ID nMyAddr = fd->_node_pool->_context->_my_node_id;

      return gigi_send_object (fd, GIGI_TYPE_ID, &nMyAddr,0);
    };
    break;
  }

  return GIGI_OK;
}

/* function _gigi_clean_invalid_socket
 *
 */
static void _gigi_clean_invalid_socket (GIGIONE context)
{
  int hash_ind;
  fd_pool * pI,* pEND;
  GIGI_SOCKET * sI, sDel = NULL;
  int nDelCount = 0, now;

  if ( (context->_idle_count % 3 ) != 0 )
  {
    return;
  }
  
  now = (int)time(NULL);
  
  MUTEX_LOCK(context->_lock);

  for (hash_ind = 0; hash_ind < NODE_HASH_COUNT; hash_ind++)
  for (pI = RING_BEGIN(context->_fd_pool_hash + hash_ind,fd_pool*),
      pEND=RING_END(context->_fd_pool_hash + hash_ind, fd_pool*); 
      pI != pEND; RING_NEXT (pI) )
    {
      for (sI = &(pI->_peer_socket); *sI;)
      {
        GIGI_SOCKET s = *sI;
      
        if( (now - s->_last_used_time) > context->_conn_expried_time)
        {
          LIST_PP_DEL(sI);
          LIST_PUSH_FRONT (sDel,s);
          nDelCount++;
        }
        else
        {
          LIST_PP_NEXT(sI);
        }
      }
    }

  context->_conn_count -= nDelCount;
  MUTEX_UNLOCK(context->_lock);

  _socket_list_free( sDel );

  if(nDelCount)
  {
    _gigi_callback (context, GIGI_CB_CLEAN_TIMEOUT_CONN, nDelCount, 0);
  }
}


